package com.bjtu.boatmq.broker.service.processor;

import com.bjtu.boatmq.broker.config.BrokerConfig;
import com.bjtu.boatmq.broker.service.outer.BrokerLogger;
import com.bjtu.boatmq.broker.service.request.RequestService;
import com.bjtu.boatmq.broker.service.thread.ThreadFactoryImpl;
import com.bjtu.boatmq.common.NettyServer;
import com.bjtu.boatmq.common.protocol.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 网络线程处理类
 * @author cmy
 * 错误编码：10010~10019
 */
public class ProcessorServiceImpl implements ProcessorService {
    // 当前服务运行状态，默认未初始化
    private RunStatus currentStatus = RunStatus.NO_INIT_STATUS;
    /* 配置变量 */
    private int brokerId;                       // broker的id
    private int brokerListenPort;               // Broker监听的端口号
    private int threadPoolSize;                 // 处理线程个数
    private int threadPollQueueSize;            // 请求队列大小
    /* 线程池 */
    private ExecutorService pullMessageExecutor;    // 消息拉取线程池
    private ExecutorService OffsetCommitExecutor;   // 偏移量提交线程池
    private Map<String, ExecutorService> sendSingleExecutorSync; // 消息发送线程池

    private final BlockingQueue<Runnable> pullThreadPoolQueue;
    private final BlockingQueue<Runnable> commitThreadPoolQueue;

    /* Netty相关对象 */
    private ServerBootstrap serverBootstrap;    // Netty
    private NioEventLoopGroup boos;             // Netty
    private NioEventLoopGroup worker;           // Netty
    private NettyServer nettyServer;            // Netty server(common模块)

    /* 请求处理服务 */
    private RequestService requestService;

    public ProcessorServiceImpl() {
        // 设置网络线程配置项
        this.brokerId = BrokerConfig.getBrokerId();
        this.brokerListenPort = BrokerConfig.getBrokerPort();
        this.threadPoolSize = BrokerConfig.getNumNetWorkThreads();
        this.threadPollQueueSize = BrokerConfig.getNumNetRequestQueues();
        // 若线程池数量小于1，则更改为"16 + 2*当前处理器数"
        if (threadPoolSize < 1) {
            threadPoolSize = 16 + Runtime.getRuntime().availableProcessors() * 2;
        }
        // 若请求队列小于0，则更改为0
        if(threadPollQueueSize < 0){
            threadPollQueueSize = 0;
        }
        // 初始化线程池
        this.pullThreadPoolQueue = new LinkedBlockingQueue<Runnable>(threadPollQueueSize);
        this.commitThreadPoolQueue = new LinkedBlockingQueue<Runnable>(threadPollQueueSize);
        this.sendSingleExecutorSync = new HashMap<>();
    }

    @Override
    public boolean initialize() {
        // 消息拉取线程池
        this.pullMessageExecutor = new ThreadPoolExecutor(
                threadPoolSize,         // 核心线程池大小
                threadPoolSize,         // 线程池最大线程数
                1000 * 60, // 空任务存活时间
                TimeUnit.MILLISECONDS,  // 时间单位
                this.pullThreadPoolQueue,// 等待队列
                new ThreadFactoryImpl("PullMessageThread_"));
        // 偏移量提交线程池
        this.OffsetCommitExecutor = new ThreadPoolExecutor(
                threadPoolSize,
                threadPoolSize,
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.commitThreadPoolQueue,
                new ThreadFactoryImpl("OffsetCommitThread_"));
        // 初始化Netty
        serverBootstrap = new ServerBootstrap();
        boos = new NioEventLoopGroup();
        worker = new NioEventLoopGroup();
        // 初始化请求处理服务
        requestService = new RequestService();
        // 更新状态为INIT_STATUS（完成初始化）
        this.setCurrentStatus(RunStatus.INIT_STATUS);
        return true;
    }

    /**
     * 启动网络服务，监听端口
     */
    @Override
    public boolean start() {
        if(getCurrentStatus() != RunStatus.INIT_STATUS ){
            BrokerLogger.write("10010","have not initialized, " +
                    "can't start processor service.");
            return false;
        }
        try {
            // 自己测试的监听
            serverBootstrap
                    .group(boos, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        protected void initChannel(NioSocketChannel ch) {
                            ch.pipeline().addLast(new SourceDecoder());
                            ch.pipeline().addLast(new SourceEncoder());
                            // ch.pipeline().addLast(new RequestHandle());
                            ch.pipeline().addLast(new SimpleChannelInboundHandler<PacketSource>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, PacketSource ps) {
                                    dealWithRequest(ctx, ps);
                                }
                            });
                        }
                    })
                    .bind(getBrokerListenPort());
        } catch (Exception e) {
            BrokerLogger.write(e);
            System.out.println("网络服务发生意外错误");
            return false;
        }
        // 更新状态为RUN_STATUS（运行状态）
        this.setCurrentStatus(RunStatus.RUN_STATUS);
        return true;
    }

    public void dealWithRequest(ChannelHandlerContext ctx, PacketSource ps) {
        if(getCurrentStatus() != RunStatus.RUN_STATUS){
            System.out.println("已关闭请求处理");
            return;
        }
        System.out.println("收到一条消息:" + ps.getRequestId() + " 来自" + ctx.channel().remoteAddress());
        // 根据请求类型插入不同的线程池，完成处理
        switch (ps.getCode()) {
            case RequestCode.SEND_MESSAGE:  // 发送消息请求
                sendMessageRequestSync(ctx, ps);
                break;
            case RequestCode.PULL_MESSAGE:  // 拉取消息请求
                pullMessageExecutor.execute(() -> {
                    requestService.pullMessageRequest(ctx, ps);
                });
                break;
            case RequestCode.COMMIT_CONSUME_OFFSET: // 偏移量提交请求
                OffsetCommitExecutor.execute(() -> {
                    requestService.commitOffsetRequest(ctx, ps);
                });
                break;
            default:    //接收到了未知的请求
                requestService.unknowRequest(ctx, ps);
                break;
        }
    }

    /**
     * 发送消息同步线程池（同分区顺序存储，不同分区无影响）
     */
    public void sendMessageRequestSync(ChannelHandlerContext ctx, PacketSource ps) {
        // 从数据包中获取消息（单条）
        System.out.println("Data Class:" + ps.getData().getClass());
        Message message = (Message) ps.getData();

        // 获取消息的主题分区，生成key，格式为"topicName--partitionId"
        String singleThreadKey = message.getTopic() + "--" + message.getPartitionId();
        System.out.println("singleThreadKey:" + singleThreadKey);
        // 查找主题分区的线程池（没有则创建）
        if (!sendSingleExecutorSync.containsKey(singleThreadKey)) {
            ExecutorService singleThreadPoll = Executors.newSingleThreadExecutor(
                    new ThreadFactoryImpl("SendMessageThread(" + singleThreadKey + ")_")
            );
            sendSingleExecutorSync.put(singleThreadKey, singleThreadPoll);
        }
        // 在线程池中分配新的线程
        sendSingleExecutorSync.get(singleThreadKey).execute(() -> {
            System.out.println("处理线程：" + Thread.currentThread().getName());
            requestService.sendMessageRequest(ctx, ps);
        });
    }

    @Override
    public boolean shutdown() {
        // 停止处理请求
        this.setCurrentStatus(RunStatus.WAIT_STATUS);
        // 停止线程池（处理完自动关闭）
        pullMessageExecutor.shutdown();
        OffsetCommitExecutor.shutdown();
        for(ExecutorService sendThreadPool : sendSingleExecutorSync.values()){
            sendThreadPool.shutdown();
        }
        this.setCurrentStatus(RunStatus.SHUTDOWN_STATUS);
        // good bye~
        BrokerLogger.writeInfo("Processor Good Bye~");
        return false;
    }

    /* 自定义set和get函数 */

    public RunStatus getCurrentStatus() {
        return currentStatus;
    }

    public void setCurrentStatus(RunStatus currentStatus) {
        this.currentStatus = currentStatus;
    }

    public int getBrokerId() {
        return brokerId;
    }

    public void setBrokerId(int brokerId) {
        this.brokerId = brokerId;
    }

    public int getBrokerListenPort() {
        return brokerListenPort;
    }

    public void setBrokerListenPort(int brokerListenPort) {
        this.brokerListenPort = brokerListenPort;
    }

    public int getThreadPoolSize() {
        return threadPoolSize;
    }

    public void setThreadPoolSize(int threadPoolSize) {
        this.threadPoolSize = threadPoolSize;
    }

    public int getThreadPollQueueSize() {
        return threadPollQueueSize;
    }

    public void setThreadPollQueueSize(int threadPollQueueSize) {
        this.threadPollQueueSize = threadPollQueueSize;
    }

    public ExecutorService getPullMessageExecutor() {
        return pullMessageExecutor;
    }

    public void setPullMessageExecutor(ExecutorService pullMessageExecutor) {
        this.pullMessageExecutor = pullMessageExecutor;
    }

    public ExecutorService getOffsetCommitExecutor() {
        return OffsetCommitExecutor;
    }

    public void setOffsetCommitExecutor(ExecutorService offsetCommitExecutor) {
        OffsetCommitExecutor = offsetCommitExecutor;
    }

    public Map<String, ExecutorService> getSendSingleExecutorSync() {
        return sendSingleExecutorSync;
    }

    public void setSendSingleExecutorSync(Map<String, ExecutorService> sendSingleExecutorSync) {
        this.sendSingleExecutorSync = sendSingleExecutorSync;
    }

    public BlockingQueue<Runnable> getPullThreadPoolQueue() {
        return pullThreadPoolQueue;
    }

    public BlockingQueue<Runnable> getCommitThreadPoolQueue() {
        return commitThreadPoolQueue;
    }

    public ServerBootstrap getServerBootstrap() {
        return serverBootstrap;
    }

    public void setServerBootstrap(ServerBootstrap serverBootstrap) {
        this.serverBootstrap = serverBootstrap;
    }

    public NioEventLoopGroup getBoos() {
        return boos;
    }

    public void setBoos(NioEventLoopGroup boos) {
        this.boos = boos;
    }

    public NioEventLoopGroup getWorker() {
        return worker;
    }

    public void setWorker(NioEventLoopGroup worker) {
        this.worker = worker;
    }

    public NettyServer getNettyServer() {
        return nettyServer;
    }

    public void setNettyServer(NettyServer nettyServer) {
        this.nettyServer = nettyServer;
    }

    public RequestService getRequestService() {
        return requestService;
    }

    public void setRequestService(RequestService requestService) {
        this.requestService = requestService;
    }
}
